package ar.edu.itba.pod.legajo45129.simul.communication;

import java.io.Serializable;
import java.rmi.RemoteException;
import java.rmi.server.UnicastRemoteObject;
import java.util.Timer;
import java.util.TimerTask;

import org.apache.log4j.Logger;

import ar.edu.itba.pod.legajo45129.simul.communication.payload.ResourceTransferMessagePayloadImpl;
import ar.edu.itba.pod.simul.communication.ConnectionManager;
import ar.edu.itba.pod.simul.communication.ThreePhaseCommit;
import ar.edu.itba.pod.simul.communication.Transactionable;
import ar.edu.itba.pod.simul.communication.payload.Payload;
import ar.edu.itba.pod.simul.market.Resource;

import com.google.common.base.Preconditions;

public class TransactionManager implements Transactionable, Serializable {
	/**
	 * 
	 */
	private static final long serialVersionUID = 3684742347429148341L;
	
	private ConnectionManagerImpl connectionManager;
	private TransactionContext transactionContext;
	private ThreePhaseCommit threepc;

	private Timer timer;

//	private long TIME_OUT;

	private static final long TIME_OUT_3PC = 5000;

	private static final long TIME_WAIT = 10000;
	
	private static Logger logger = Logger.getLogger(TransactionContext.class);

	public TransactionManager(ConnectionManagerImpl connectionManagerImpl) throws RemoteException {
		UnicastRemoteObject.exportObject(this, 0);
		this.connectionManager = connectionManagerImpl;
		threepc = new ThreePhaseCommitImpl(connectionManager);
	}

	@Override
	public void acceptTransaction(String remoteNodeId) throws RemoteException {
		if (transactionContext != null && transactionContext.isBusy()) {
			try {
				Thread.sleep(TIME_WAIT);
			} catch (InterruptedException e) {
				logger.fatal(e);
				throw new RuntimeException(e.getMessage());
			}
			
			if (transactionContext.isBusy()) {
				throw new IllegalStateException("transaction context busy");
			}
		}
		
		transactionContext = new TransactionContext(remoteNodeId, connectionManager.getNodeId().toString());
		logger.debug("transaction context accepted");

	}

	@Override
	public void beginTransaction(String remoteNodeId, long timeout)
			throws RemoteException {
		Preconditions.checkNotNull(remoteNodeId);
		String localNode = connectionManager.getNodeId().toString();
		
		if (transactionContext != null && transactionContext.isBusy()) {
			throw new IllegalStateException("trying to create a transaction that already exist");
		}

		logger.debug("request transaction context");
		connectionManager.getConnectionManager(remoteNodeId).getNodeCommunication().acceptTransaction(localNode);
		transactionContext = new TransactionContext(localNode, remoteNodeId);
		
		timer = new Timer();
		timer.schedule(transactionContext, timeout);
		
		logger.debug("transaction context accepted");
		
	}

	@Override
	public void endTransaction() throws RemoteException {
		if (transactionContext == null) {
			throw new IllegalStateException("no transaction exist");
		}
		
		if (!transactionContext.isBusy()) {
			throw new IllegalStateException("transaction does not exist");
		}
		
		if (connectionManager.getNodeId().toString().equals(transactionContext.getSource())) {
			logger.debug("aplying exchange operation");
			logger.debug("begin three phase commit");
			doThreePhaseCommit();
			ConnectionManager remoteCm = connectionManager.getConnectionManager(transactionContext.getDest());
			logger.debug("ending remote transaction context");
			remoteCm.getNodeCommunication().endTransaction();
			logger.debug("three phase commit done");
		} else {
			logger.debug("remote transaction context ended");
		}
		
		
		synchronized (transactionContext) {
			// cancel the timer
			transactionContext.cancel();
			transactionContext = null;
		}
		
		logger.debug("transaction finished");
		
	}

	private void doThreePhaseCommit() throws RemoteException {
		String coordinator = connectionManager.getNodeId().toString();
		ThreePhaseCommit remoteThreePhaseCommit = connectionManager
													.getConnectionManager(transactionContext.dest)
													.getThreePhaseCommit();
		if (!remoteThreePhaseCommit.canCommit(coordinator, TIME_OUT_3PC)) {
			logger.error("remote can-commit phase error, aborting");
			remoteThreePhaseCommit.abort();
			threepc.abort();
			return;
		}
		
		if (!threepc.canCommit(coordinator, TIME_OUT_3PC)) {
			logger.error("local can-commit phase error, aborting");
//			threepc.abort();
			return;
		}
		logger.debug("can-commit phase done");
		
		try {
			remoteThreePhaseCommit.preCommit(coordinator);
		} catch (Exception e) {
			logger.equals("remote pre-commit phase error, aborting");
			remoteThreePhaseCommit.abort();
			threepc.abort();
			return;
		}
		
		try {
			threepc.preCommit(coordinator);
		} catch (Exception e) {
			logger.equals("local pre-commit phase error, aborting");
			threepc.abort();
			return;
		}		
		logger.debug("pre-commit phase done");
		
		try {
			remoteThreePhaseCommit.doCommit(coordinator);
		} catch (Exception e) {
			logger.equals("remote do-commit phase error, aborting");
			remoteThreePhaseCommit.abort();
			threepc.abort();
			return;
		}
		
		try {
			threepc.doCommit(coordinator);
		} catch (Exception e) {
			logger.equals("local do-commit phase error, aborting");
			threepc.abort();
			return;
		}
		
		logger.debug("do-commit phase done");
	}

	@Override
	public void exchange(Resource resource, int amount, String sourceNode,
			String destinationNode) throws RemoteException {
		Preconditions.checkNotNull(resource);
		Preconditions.checkNotNull(sourceNode);
		Preconditions.checkNotNull(destinationNode);
		Preconditions.checkArgument(amount > 0);
		
		ExchangeInfo exchangeInfo = new ExchangeInfo(resource, amount, sourceNode, destinationNode);
		
		if (sourceNode.equals(destinationNode)) {
			throw new IllegalStateException("source node must be different to destination node");
		}
		
		if (!transactionContext.isNodeBelongs(sourceNode)) {
			throw new IllegalStateException(sourceNode + " does not belong to the transaction");
			
		}
		
		if (!transactionContext.isNodeBelongs(destinationNode)) {
			throw new IllegalStateException(destinationNode + " does not belong to the transaction");
		}
		
		transactionContext.setExchange(exchangeInfo);
		// if localnode is equal to source node of the exchange, then i must set
		// exchange on the destination node, so both side known that the exchange has
		// be done.
		if (connectionManager.getNodeId().equals(sourceNode)) {
			connectionManager.getConnectionManager(destinationNode)
								.getNodeCommunication()
								.exchange(resource, amount, sourceNode, destinationNode);
		}
		
		transactionContext.setDone(true);
		logger.debug("exchange done");
	}

	@Override
	public Payload getPayload() throws RemoteException {
		if (!transactionContext.isBusy()) {
			throw new IllegalStateException("no transaction context running");
		}
		
		if (!transactionContext.isDone()) {
			throw new IllegalStateException("transaction not finished yet");
		}
		
		return new ResourceTransferMessagePayloadImpl(transactionContext.getExchangeInfo()) ;
	}

	@Override
	public void rollback() throws RemoteException {		
		Preconditions.checkNotNull(transactionContext);
		
		transactionContext.cancel();
		
		if (transactionContext.isDone()) {
			logger.error("transaction already done");
			throw new IllegalStateException("transaction already done");
		}
		

		if (connectionManager.getNodeId().toString()
				.equals(transactionContext.getExchangeInfo().getSourceNode())) {
			ConnectionManager remoteCM = connectionManager.getConnectionManager(transactionContext.getExchangeInfo().getDestinationNode());
			remoteCM.getNodeCommunication().rollback();
		}
		
		threepc.abort();
		
		synchronized (transactionContext) {			
			transactionContext = null;
		}
		
		logger.debug("roll back done");
	}
	
	private class TransactionContext extends TimerTask {
		private boolean busy = false;
		private String source;
		private String dest;
		private boolean done = false;
		private ExchangeInfo exchangeInfo;

		public TransactionContext(String source, String dest) {
			this.source = source;
			this.dest = dest;
			busy = true;
//			new Timer().schedule(this, timeout);
		}

		public String getDest() {
			return dest;
		}

		public String getSource() {
			return source;
		}
		
		public synchronized ExchangeInfo getExchangeInfo() {
			return exchangeInfo;
		}

		public void setDone(boolean value) {
			done = value;
			
		}
		
		public synchronized boolean isDone() {
			return done;
		}

		/**
		 * Set the exchange information
		 * @param resource
		 * @param amount
		 * @param sourceNode
		 * @param destinationNode
		 */
		public synchronized void setExchange(ExchangeInfo exchangeInfo) {
			this.exchangeInfo = exchangeInfo;
			
		}

		/**
		 * Check if given node belongs to the transaction context. 
		 * @param node
		 * @return true if node is the local node or the remote node of this transaction context.
		 */
		public synchronized boolean isNodeBelongs(String node) {
			Preconditions.checkNotNull(node);
			Preconditions.checkNotNull(source);
			Preconditions.checkNotNull(dest);

			if (source.equals(node)) {
				return true;
			}
			
			if (dest.equals(node)) {
				return true;
			}
			
			return false;
		}

		/**
		 * Check if the transaction is currently in use.
		 * @return true if the transaction is in use.
		 */
		public synchronized boolean isBusy() {
			return busy;
		}


		private void onTimeout(){
			if (!transactionContext.isDone()) {
				logger.warn("transaction time out, taking roll back action");
				try {
					rollback();
				} catch (RemoteException e) {
					logger.error("error in time out rolling back process");
					throw new RuntimeException(e.getMessage());
				}
			}
			
		}
		

		@Override
		public void run() {
			onTimeout();
		}		
	}

	public ExchangeInfo getExchangeInfo() {
		Preconditions.checkNotNull(transactionContext);
		return transactionContext.getExchangeInfo();
	}

	public ThreePhaseCommit getThreepc() {
		return threepc;
	}
}

